package com.interview.rpc.client;

import com.interview.rpc.codec.RpcDecoder;
import com.interview.rpc.codec.RpcEncoder;
import com.interview.rpc.protocol.Request;
import com.interview.rpc.protocol.Response;
import com.interview.rpc.serialize.JsonSerializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * RPC客户端
 * 
 * 使用Netty实现的RPC客户端，负责与RPC服务器建立连接并发送请求
 */
public class RpcClient {
    
    // 服务器地址
    private String host;
    // 服务器端口
    private int port;
    // Netty客户端启动引导类
    private Bootstrap bootstrap;
    // Netty事件循环组
    private EventLoopGroup group;
    // 与服务器的连接通道
    private Channel channel;
    // 存储正在进行的请求，key为请求ID，value为CompletableFuture对象
    private static ConcurrentHashMap<String, CompletableFuture<Response>> pendingRequests = new ConcurrentHashMap<>();
    
    public RpcClient(String host, int port) {
        this.host = host;
        this.port = port;
    }
    
    /**
     * 连接到RPC服务器
     * 
     * @throws InterruptedException
     */
    public void connect() throws InterruptedException {
        // 创建事件循环组
        group = new NioEventLoopGroup();
        // 创建客户端启动引导类
        bootstrap = new Bootstrap();
        bootstrap.group(group)
                // 指定通道类型为NioSocketChannel
                .channel(NioSocketChannel.class)
                // 设置处理器
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        // 添加解码器，用于将字节流解码为Response对象
                        pipeline.addLast(new RpcDecoder(Response.class, new JsonSerializer()));
                        // 添加编码器，用于将Request对象编码为字节流
                        pipeline.addLast(new RpcEncoder(Request.class, new JsonSerializer()));
                        // 添加客户端处理器，用于处理服务器响应
                        pipeline.addLast(new RpcClientHandler(pendingRequests));
                    }
                });
        
        // 连接到服务器并获取通道
        channel = bootstrap.connect(host, port).sync().channel();
    }
    
    /**
     * 发送RPC请求并等待响应
     * 
     * @param request RPC请求对象
     * @return RPC响应对象
     * @throws Exception
     */
    public Response sendRequest(Request request) throws Exception {
        // 创建CompletableFuture对象用于异步等待响应
        CompletableFuture<Response> future = new CompletableFuture<>();
        // 将请求ID和CompletableFuture对象存入映射中
        pendingRequests.put(request.getRequestId(), future);
        
        // 发送请求
        channel.writeAndFlush(request);
        
        // 等待响应，设置超时时间为5秒
        return future.get(5, TimeUnit.SECONDS);
    }
    
    /**
     * 关闭客户端连接
     */
    public void close() {
        if (channel != null) {
            channel.close();
        }
        if (group != null) {
            group.shutdownGracefully();
        }
    }
}